package com.manbu.live.user.provider.config;

import jakarta.annotation.Resource;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.checkerframework.checker.units.qual.C;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.*;

/**
 * @author hanWang
 * @Title: :Rocketmq  生产者 配置类
 * @Package
 * @Description:
 * @date 2024/1/243:07 AM
 */

@Configuration
public class RocketMQProducerConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(RocketMQProducerConfig.class);
    @Resource
    private RocketMQProducerProperties rocketMQProducerProperties;

    @Value("${spring.application.name}")
    private String applicationName;
    /**
     * 虽然我们使用mq的时候是使用DefaultMQProducer 这个对象  但是 DefaultMQProducer 实现了MQProducer 接口
     * 在 MQProducer 接口中已经有了很多send的方法
     * @return
     */
    @Bean
    public MQProducer mqProducer(){
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(100, 150, 3, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
            // 线程池里面有很多工作线程执行 生产工作线程 实际线程池生产线程的发源地
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                // 重写名字好处  在linux环境中可以查看哪个线程 阻塞了 方便看哪个业务线程出问题
                thread.setName(applicationName+":rmq-producer:"+ThreadLocalRandom.current().nextInt(1000));
                return null;
            }
        });
        //ExecutorService executorService = Executors.newCachedThreadPool();
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
        try {

        defaultMQProducer.setNamesrvAddr(rocketMQProducerProperties.getNameSrv());
        defaultMQProducer.setProducerGroup(rocketMQProducerProperties.getGroupName());
        // 重试次数
        defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProducerProperties.getRetryTimes());
        defaultMQProducer.setRetryTimesWhenSendAsyncFailed(rocketMQProducerProperties.getRetryTimes());
        //broker 存储失败是否从新存储
        defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        //设置异步发送的 线程池
        defaultMQProducer.setAsyncSenderExecutor(threadPoolExecutor);
            defaultMQProducer.start();
            LOGGER.info("mq 生产者 启动成功，namesrv is {}", rocketMQProducerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }
        return defaultMQProducer;

    }

}
